package hk.linkcredit.common.rocketmq.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 消息发送方
 */
@Slf4j
@Component
public class MqProducer {

    public static final String BEAN_NAME = "beanName";
    public static final String MSG_ID = "msgId";
    public static final String BIZ_ID = "bizId";

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 普通字符串消息
     */
    public void sendMessage() {
        String json = "普通消息";
        rocketMQTemplate.convertAndSend("sendMessage", json);
    }

    /**
     * 发送同步消息（阻塞当前线程，等待broker响应发送结果，这样不太容易丢失消息）
     * sendResult为返回的发送结果
     */
    public SendResult syncSend(String topic, Object msg) {
        String json = "同步消息";
        SendResult sendMessage = rocketMQTemplate.syncSend(topic, msg);
        System.out.println(sendMessage);
        return sendMessage;
    }

    /**
     * 发送带tag的消息，直接在topic后面加上":tag"
     * sendResult为返回的发送结果
     *
     * @param topic
     * @param tag   消息tag
     * @param msg
     * @return
     */
    public SendResult syncSendWithTag(String topic, String tag, Object msg) {
        topic = topic + ":" + tag;
        log.info("开始发送MQ[tag={}]消息 >>> topic = {}, Msg = {}", tag, topic, msg.toString());
        SendResult sendMessage = rocketMQTemplate.syncSend(topic, msg);
        log.info("结束发送MQ[tag={}]消息 >>> [消息发送结果]：{}", tag, sendMessage.toString());
        return sendMessage;
    }

    /**
     * 异步消息
     */
    public void asyncSend() {
        String json = "异步消息";
        SendCallback callback = new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("123");
            }

            @Override
            public void onException(Throwable throwable) {
                System.out.println("456");
            }
        };
        rocketMQTemplate.asyncSend("sendMessage", json, callback);
    }

    /**
     * 单向消息
     */
    public void onewaySend() {
        String json = "单向消息";
        rocketMQTemplate.sendOneWay("sendMessage", json);
    }

    /**
     * 发送事务消息
     *
     * @param topic    主题+tag
     * @param msg      消息内容
     * @param msgId    消息唯一id
     * @param bizId    业务唯一id
     * @param beanName 实现了ITransactionHandler接口的beanName，spring默认为类名并且首字母小写
     * @param obj      事务处理所需的参数
     * @return
     */
    public TransactionSendResult sendMessageInTransaction(String topic, String msg, String msgId, String bizId, String beanName, Object obj) {
        assert StringUtils.isNotEmpty(topic);
        assert StringUtils.isNotEmpty(msg);
        assert StringUtils.isNotEmpty(msgId);
        assert StringUtils.isNotEmpty(bizId);
        assert StringUtils.isNotEmpty(beanName);
        assert obj != null;
        return rocketMQTemplate.sendMessageInTransaction(
                topic,
                MessageBuilder.withPayload(msg)
                        .setHeader(BEAN_NAME, beanName)
                        .setHeader(MSG_ID, msgId)
                        .setHeader(BIZ_ID, bizId)
                        .build(),
                obj
        );
    }
}

